package com.kwan.xframe.api;

import android.util.Log;



import java.net.InetSocketAddress;
import java.net.Socket;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import okio.BufferedSink;
import okio.BufferedSource;
import okio.Okio;

public class RxSocket {

	private Socket socket;
	private BufferedSink mSink;
	private BufferedSource mSource;
	private String ip;
	private int port;

	//"192.168.0.26", 12000
	public RxSocket(String ip, int port) {
		this.ip = ip;
		this.port = port;
	}

	public Flowable<byte[]> write(final byte[] data) {
		return Flowable
				.create(new FlowableOnSubscribe<byte[]>() {
					@Override
					public void subscribe(FlowableEmitter<byte[]> e) throws Exception {

						if (socket != null && socket.isConnected()) {
							Log.e("kwan", "socket is connect now 2 close");
							socket.close();
						}

						socket = new Socket();
						socket.connect(new InetSocketAddress(ip, port));
						Log.e("kwan", "isConnected " + socket.isConnected());

						mSink = Okio.buffer(Okio.sink(socket));
						mSource = Okio.buffer(Okio.source(socket));

						mSink.write(data);
						mSink.flush();



						byte[] buff = new byte[2];
						byte data[] = new byte[]{};

						for (int i = 0;i<2;i++){
							Log.e("kwan", "xxx -- " );
							Log.e("kwan", "read -- " + mSource.read(buff));
						//	data = ByteUtil.concatAll(data,buff);
						}

					//	short slen = ByteUtil.getShort(data,2);

					//	Log.e("kwan", "data slen read -- " + slen);
						Log.e("kwan", "data read -- " + data.length);

					//	byte[] buff2 = new byte[slen];
					//	mSource.read(buff2);

						//data = ByteUtil.concatAll(data,buff2);
						//mSource.re
						Log.e("kwan", "data read -- " + data.length);

						e.onNext(data);
						e.onComplete();
						mSink.close();
						mSource.close();
						socket.close();
					}
				}, BackpressureStrategy.BUFFER);

	}
}
